package mylab.io.base.context;

import com.google.common.collect.Maps;
import com.google.common.util.concurrent.ListenableScheduledFuture;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import mylab.base.foundation.event.EventUtil;
import mylab.base.foundation.lang.Callback;
import mylab.io.base.Das;
import mylab.io.base.cache.DasCache;
import mylab.io.base.cache.DefaultDasCache;
import mylab.io.base.event.DasExceptionEvent;
import mylab.io.base.event.DasHeartbeatEvent;
import mylab.io.base.event.DasStatusChangedEvent;
import mylab.io.base.event.MsgReceivedEvent;
import mylab.io.base.event.TerminalConnectionChangedEvent;
import mylab.io.base.fsm.DasStatus;
import mylab.io.base.model.Msg;
import mylab.io.base.model.Terminal;
import mylab.io.base.property.CommonAttrs;
import mylab.io.base.property.DasProperties;
import mylab.io.base.sync.Synchro;
import mylab.io.base.util.DasUtil;
import mylab.utils.common.concurrent.RunnableUtil;

@Slf4j
public class DefaultDC<P extends DasProperties, M extends Msg> implements DasContext<P, M> {

    @Getter
    protected Das<?, P, M> das;
    @Getter
    protected P dasPropertites;
    @Getter
    protected DasCache dasCache;
    @Getter
    protected Synchro synchro;

    protected Thread hook = new Thread(this::dispose);
    protected ListenableScheduledFuture<?> heartbeatFuture;
    protected ListenableScheduledFuture<?> daemonFuture;
    protected Map<String, AtomicBoolean> terminalConnectionMap = Maps.newConcurrentMap();

    public DefaultDC(P dasPropertites) {
        this.dasPropertites = dasPropertites;
    }

    @Override
    public void init(@NonNull Das<?, P, M> das) {
        DasContext.super.init(das);
        this.das = das;

        //VM 挂钩：当进程正常退出时清理
        Runtime.getRuntime().addShutdownHook(hook);
        this.dasCache = new DefaultDasCache();
        this.terminalConnectionMap = Maps.newConcurrentMap();

        CommonAttrs commonAttrs = getDasPropertites().getCommon();
        if (commonAttrs != null && commonAttrs.getSync() != null) {
            this.synchro = commonAttrs.getSync().getSynchro();
            scheduleDaemon();
            scheduleHeartbeat();
        }
    }

    @Override
    public void dispose() {
        if (Objects.equals(das.getStatus(), DasStatus.Running)) {
            das.stop(null);
        }

        cancelDaemon();
        cancelHeartbeat();

        this.terminalConnectionMap.clear();
        if (this.synchro != null) {
            this.synchro.clear();
        }

        this.synchro = null;
        this.dasCache = null;

        DasContext.super.dispose();
    }

    //------------------------------------ 定时任务  ----------------------------

    protected void scheduleHeartbeat() {
        if (dasPropertites.getCommon().isHeartbeat()) {
            heartbeatFuture = RunnableUtil.scheduleAtFixedRate(this::heartbeatFunc, 1, dasPropertites.getCommon().getHeartbeatSeconds(), TimeUnit.SECONDS);
        }
    }

    protected void cancelHeartbeat() {
        if (heartbeatFuture != null) {
            heartbeatFuture.cancel(true);
        }
        heartbeatFuture = null;

    }

    protected void heartbeatFunc() {
        //只有正常运行态才会正常心跳
        if (Objects.equals(DasStatus.Running, das.getStatus())) {
            log.debug("### heartbeat ### das id: {}", das.getId());

            //缓存正常的das, 在查询缓存时会自动清理过期的das
            DasProperties dasInfo = das.getContext().getDasPropertites();
            dasCache.addDasInfo(dasInfo);

            //post event
            if (dasPropertites.getCommon().isEventEnabled()) {
                EventUtil.postAsyncEvent(new DasHeartbeatEvent(das, null));
            }
        }
    }

    protected void scheduleDaemon() {
        if (dasPropertites.getCommon().isDaemon()) {
            daemonFuture = RunnableUtil.scheduleWithFixedDelay(this::daemonFunc, 10, dasPropertites.getCommon().getDaemonSeconds(), TimeUnit.SECONDS);
        }
    }

    protected void cancelDaemon() {
        if (daemonFuture != null) {
            daemonFuture.cancel(true);
        }
        daemonFuture = null;
    }

    protected void daemonFunc() {
        if (das == null || das.getStatus() == null) {
            return;
        }

        if (!Objects.equals(das.getStatus(), DasStatus.Running)) {
            log.debug("### restart ### current status:{}", das.getStatus());
            log.info("### daemon ### 开始执行");


            das.start(new Callback<Void>() {
                @Override
                public void onSuccess(Void unused) {
                    log.info("### daemon ### 成功");
                }

                @Override
                public void onFailure(String cause) {
                    log.warn("### daemon ### 失败:{}", cause);
                }

                @Override
                public void onException(Throwable throwable) {
                    log.error("### daemon ### 异常:{}", throwable.getMessage());
                }
            });
        }

    }

    //------------------------------------ DasEventCallback  ----------------------------

    @Override
    public void onDasStatusChanged(DasStatus status) {
        DasUtil.logStatus(das, status);

        //cache
        DasProperties dasInfo = das.getContext().getDasPropertites();
        boolean isRunning = Objects.equals(DasStatus.Running, status);
        if (isRunning) {
            dasCache.addDasInfo(dasInfo);
        } else {
            dasCache.removeDasInfo(das.getId());
        }

        // post event
        if (dasPropertites.getCommon().isEventEnabled()) {
            EventUtil.postAsyncEvent(new DasStatusChangedEvent(das, status));
        }
        //notify watcher
        if (notifyWatcherEnabled()) {
            DasUtil.postRunnable(() -> das.getWatcher().onDasStatusChanged(das, status));
        }
    }

    @Override
    public void onTerminalConnectionChanged(Terminal terminal, boolean isConnected) {
        String terminalId = terminal.getId();
        if (!terminalConnectionMap.containsKey(terminalId)) {
            terminalConnectionMap.put(terminalId, new AtomicBoolean(false));
        }

        AtomicBoolean atomic = terminalConnectionMap.get(terminalId);
        if (atomic.get() != isConnected) {
            DasUtil.logTerminalConnection(das, terminal, isConnected);
            atomic.set(isConnected);

            //cache terminal
            if (isConnected) {
                dasCache.addTerminal(terminal);
            } else {
                dasCache.removeTerminal(terminalId);

                if (dasPropertites.getCommon().isClientMode()) {
                    log.warn("当前das是client，唯一连接断开");
                    das.stop(null);
                }
            }

            // post event
            if (dasPropertites.getCommon().isEventEnabled()) {
                EventUtil.postAsyncEvent(new TerminalConnectionChangedEvent(terminal, isConnected));
            }
            //notify watcher
            if (notifyWatcherEnabled()) {
                DasUtil.postRunnable(() -> das.getWatcher().onTerminalConnectionChanged(das, terminal, isConnected));
            }
        }
    }

    @Override
    public void onReceived(M msg) {
        if (synchro != null && synchro.tryMatch(msg)) {
            //同步消息，此处不再处理
            DasUtil.logMsg(msg, false, true);
            return;
        }

        //非同步回应，则作为异步消息处理
        DasUtil.logMsg(msg, false, false);

        // post event
        if (dasPropertites.getCommon().isEventEnabled()) {
            EventUtil.postAsyncEvent(new MsgReceivedEvent(das, msg));
        }
        //notify watcher
        if (notifyWatcherEnabled()) {
            DasUtil.postRunnable(() -> das.getWatcher().onReceived(das, msg));
        }
    }

    @Override
    public void onException(Throwable throwable) {
        DasUtil.logException(das, throwable);

        // post event
        if (dasPropertites.getCommon().isEventEnabled()) {
            EventUtil.postAsyncEvent(new DasExceptionEvent(das, throwable));
        }
        //notify watcher
        if (notifyWatcherEnabled()) {
            DasUtil.postRunnable(() -> das.getWatcher().onException(das, throwable));
        }
    }

    protected boolean notifyWatcherEnabled() {
        return dasPropertites.getCommon().isWatcherEnabled() && das.getWatcher() != null;
    }

}
